
package org.springblade.modules.api;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.github.xiaoymin.knife4j.annotations.ApiOperationSupport;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam;
import lombok.AllArgsConstructor;
import org.springblade.common.constant.CommonConstant;
import org.springblade.core.boot.ctrl.BladeController;
import org.springblade.core.cache.utils.CacheUtil;
import org.springblade.core.launch.constant.AppConstant;
import org.springblade.core.mp.support.Condition;
import org.springblade.core.mp.support.Query;
import org.springblade.core.tenant.annotation.NonDS;
import org.springblade.core.tool.api.R;
import org.springblade.core.tool.jackson.JsonUtil;
import org.springblade.core.tool.utils.DateUtil;
import org.springblade.core.tool.utils.Func;
import org.springblade.core.tool.utils.WebUtil;
import org.springblade.modules.flume.entity.TestFlumeData;
import org.springblade.modules.flume.service.ITestFlumeDataService;
import org.springblade.modules.kafka.constant.KafkaConstant;
import org.springblade.modules.system.entity.ApiScope;
import org.springblade.modules.system.service.IApiScopeService;
import org.springblade.modules.system.vo.ApiScopeVO;
import org.springblade.modules.system.wrapper.ApiScopeWrapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.*;
import sun.misc.BASE64Decoder;

import javax.validation.Valid;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

import static org.springblade.core.cache.constant.CacheConstant.SYS_CACHE;

/**
 *
 *
 * @author
 */
@NonDS
@RestController
@AllArgsConstructor
@RequestMapping(CommonConstant.GATHER_API_NAME + "")
@Api(value = "", tags = "")
public class KafkaApiController extends BladeController {

	@Autowired
	private KafkaTemplate<String, String> kafkaTemplate;
	@Autowired
	private ITestFlumeDataService testFlumeDataService;

	/**
	 *
	 */
//	@GetMapping("/send")
//	@ApiOperationSupport(order = 1)
//	@ApiOperation(value = "", notes = "")
//	public R send(@RequestParam("data") String  data) {
//		BASE64Decoder decoder = new BASE64Decoder();
//		try {
//			String ds=new String(decoder.decodeBuffer(data), "UTF-8");
//			kafkaTemplate.send(KafkaConstant.TOPIC_BURIED, ds);
//		} catch (IOException e) {
//			e.printStackTrace();
//			return R.status(false);
//		}
//
//		return R.status(true);
//	}

	@RequestMapping("/send")
	@ApiOperationSupport(order = 1)
	@ApiOperation(value = "", notes = "")
	public R send(@RequestParam(value="data_list",required=false) String dataList
		,@RequestParam(value="data",required=false) String data
		,@RequestParam(value="gzip",required=false) String  gzip) {

		if (true){
//			String mes = "{\"name\":\"test\"}";
//			kafkaTemplate.send(KafkaConstant.TOPIC_GATHER, mes);
			return R.status(true);
		}
		BASE64Decoder decoder = new BASE64Decoder();
		try {
			String predata = Func.isEmpty(dataList)?data:dataList;
			String userAgent = WebUtil.getHeader("User-Agent");
			if (Func.isNotEmpty(userAgent)){
				if (userAgent.contains("ios") || userAgent.contains("iOS") || userAgent.contains("IOS")){
					predata = java.net.URLDecoder.decode(predata, "utf-8");
				}
			}
			if ("1".equals(gzip)){
				predata = GZipUtil.uncompress(predata);
			}else{
				predata=new String(decoder.decodeBuffer(predata), "UTF-8");
			}
			String ds=predata;
//			String ds=new String(decoder.decodeBuffer(predata), "UTF-8");
//			if ("1".equals(gzip)){
//				ds = GZipUtil.uncompress(ds);
//			}

			kafkaTemplate.send(KafkaConstant.TOPIC_GATHER, ds);

//					TestFlumeData flumeData = new TestFlumeData();
//		flumeData.setGatherData(ds);
//		flumeData.setCreateTime(DateUtil.now());
//		testFlumeDataService.save(flumeData);
		} catch (Exception e) {
			e.printStackTrace();
			return R.status(false);
		}

		return R.status(true);
	}

	@RequestMapping("/sendPost")
	@ApiOperationSupport(order = 1)
	@ApiOperation(value = "", notes = "")
	public R sendPost(@RequestParam(value="data_list",required=false) String dataList
				 ,@RequestParam(value="data",required=false) String data
		         ,@RequestParam(value="gzip",required=false) String  gzip) {
		BASE64Decoder decoder = new BASE64Decoder();
		try {
			String predata = Func.isEmpty(dataList)?data:dataList;
			String userAgent = WebUtil.getHeader("User-Agent");
			if (Func.isNotEmpty(userAgent)){
				if (userAgent.contains("ios") || userAgent.contains("iOS") || userAgent.contains("IOS")
				|| userAgent.contains("iPhone") || userAgent.contains("IPHONE") || userAgent.contains("iphone")
				|| userAgent.contains("iPad") || userAgent.contains("IPAD") || userAgent.contains("ipad")){
					predata = java.net.URLDecoder.decode(predata, "utf-8");
				}
			}
			if ("1".equals(gzip)){
				predata = GZipUtil.uncompress(predata);
			}else{
				predata=new String(decoder.decodeBuffer(predata), "UTF-8");
			}
			String ds=predata;
//			String ds=new String(decoder.decodeBuffer(predata), "UTF-8");
//			if ("1".equals(gzip)){
//				ds = GZipUtil.uncompress(ds);
//			}

			kafkaTemplate.send(KafkaConstant.TOPIC_BURIED, ds);
		} catch (Exception e) {
			e.printStackTrace();
			return R.status(false);
		}

		return R.status(true);
	}

	/**
	 *
	 */
	@PostMapping("/send-post")
	@ApiOperationSupport(order = 1)
	@ApiOperation(value = "", notes = "")
	public R sendPost(@RequestBody JSONObject jsonObject) {

		kafkaTemplate.send(KafkaConstant.TOPIC_BURIED, jsonObject.toJSONString());
		return R.status(true);
	}

	/**
	 *
	 */
	@PostMapping("/send-list")
	@ApiOperationSupport(order = 1)
	@ApiOperation(value = "", notes = "")
	public R sendList(@RequestBody JSONObject jsonObject) {

		kafkaTemplate.send(KafkaConstant.TOPIC_BURIED, jsonObject.toJSONString());
		return R.status(true);
	}



}
